package sprs;

import scala.Tuple2;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.rosuda.REngine.REXP;
import org.rosuda.REngine.Rserve.RConnection;

import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;

public final class JavaWordCount {
	private static final Pattern SPACE = Pattern.compile(" ");

	public static void main(String[] args) throws Exception {

		args = new String[] { "/temp/test.txt" };
		if (args.length < 1) {
			// System.err.println("Usage: JavaWordCount <file>");
			// System.exit(1);
		}

		SparkConf sparkConf = new SparkConf().setAppName("JavaWordCount");
		sparkConf.setMaster("local");
		JavaSparkContext ctx = new JavaSparkContext(sparkConf);
		JavaRDD<String> lines = ctx.textFile(args[0], 1);

		JavaRDD<String> words = lines
				.flatMap(new FlatMapFunction<String, String>() {
					// @Override
					public Iterable<String> call(String s) {
						try {
							RConnection rc = RConnectionPool.getPool()
									.borrowObject();

							REXP x = rc.eval("R.version.string");
							System.out.println(x.asString());
						} catch (Exception e) {
							e.printStackTrace();
						}
						return Arrays.asList(SPACE.split(s));
					}
				});

		JavaPairRDD<String, Integer> ones = words
				.mapToPair(new PairFunction<String, String, Integer>() {
					// @Override
					public Tuple2<String, Integer> call(String s) {
						System.out.println(s);
						return new Tuple2<String, Integer>(s, 1);
					}
				});

		JavaPairRDD<String, Integer> counts = ones
				.reduceByKey(new Function2<Integer, Integer, Integer>() {
					// @Override
					public Integer call(Integer i1, Integer i2) {
						return i1 + i2;
					}
				});
		counts.saveAsTextFile("wordcount/test1/");
	}

}